package com.pw.study.flink.source

import com.pw.study.flink.entities.UserBean
import org.apache.flink.streaming.api.functions.source.SourceFunction

import scala.util.Random

class CustomSource extends SourceFunction[UserBean]{
  var isSwitch=true
  override def run(ctx: SourceFunction.SourceContext[UserBean]): Unit = {
    while(isSwitch){
      //构建自定义数据
      val uid ="uid_" +new Random().nextInt(30)
      val ts: Long = System.currentTimeMillis()
      val vc: Int = new Random().nextInt(100) + 20
      ctx.collect(UserBean(uid,ts,vc))
      println("构建一条数据："+UserBean.toString())
      Thread.sleep(100)
    }
  }

  override def cancel(): Unit = {
    isSwitch=false
  }
}
